package consumer

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"go_knowledge/mq/kafka/event"
)

type defaultConsumer struct {
	handler event.Handler
	logger  *logrus.Logger
}

func NewDefaultConsumer(handler event.Handler) *defaultConsumer {
	return &defaultConsumer{
		handler: handler,
	}
}

func (c *defaultConsumer) SetLogger(logger *logrus.Logger) {
	c.logger = logger
}

func (c *defaultConsumer) Setup(sarama.ConsumerGroupSession) error {
	// 没有日志，则初始化一个默认日志处理
	if nil == c.logger {
		l := &logrus.Logger{}
		l.SetLevel(logrus.DebugLevel)
		c.logger = l
		fmt.Println("not set logger, use default &clog.Logger{}")
	}

	return nil
}

func (c *defaultConsumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
func (c *defaultConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

	c.logger.Debug("consume claim enter")

	for msg := range claim.Messages() {
		c.logger.Debugf(
			"message topic[%q] timestamp[%v] partition[%d] offset[%d]",
			msg.Topic, msg.Timestamp, msg.Partition, msg.Offset,
		)

		// 解析
		pbMsg := &event.PbEventMessage{}
		err := json.Unmarshal(msg.Value, pbMsg)
		if nil != err {
			c.logger.Errorf("msg unmarshal failed, data: %s", string(msg.Value))
			return errors.Wrapf(err, "msg unmarshal failed")
		}

		if err := c.handler(pbMsg); err != nil {
			c.logger.Info(fmt.Sprintf("event handle failed, data: %s, err: %s", string(msg.Value), err))
			return errors.Wrapf(err, "event handle failed，data: %s", msg.Value)
		}

		session.MarkMessage(msg, "")
	}

	return nil
}
